package com.pw.study.flink.sql;

import com.pw.study.flink.entities.WaterSensor;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;

public class $05SqlBase {
    public static void main(String[] args) {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        DataStreamSource<WaterSensor> stream = env.fromElements(
                new WaterSensor("sensor_1", 1000L, 10),
                new WaterSensor("sensor_1", 2000L, 20),
                new WaterSensor("sensor_2", 3000L, 30),
                new WaterSensor("sensor_1", 4000L, 40),
                new WaterSensor("sensor_1", 5000L, 50),
                new WaterSensor("sensor_2", 6000L, 60)
        );

        Table tb = tEnv.fromDataStream(stream);

        tb.printSchema();

        tEnv.createTemporaryView("s1",tb);
        tEnv.sqlQuery("select * from s1 where id='sensor_1' ").execute().print();

        tEnv.createTemporaryView("s2",tb);
        tEnv.sqlQuery("select " +
                "id," +
                "sum(vc) sum_vc " +
                "from s2 " +
                "group by id").execute().print();


    }

}
